package cn.kuporo.seckill.order.application.listener;

import cn.kuporo.seckill.common.cache.distribute.DistributedCacheService;
import cn.kuporo.seckill.common.constants.MQConstants;
import cn.kuporo.seckill.common.constants.RedisConstants;
import cn.kuporo.seckill.common.model.message.TxMessage;
import cn.kuporo.seckill.order.application.place.SeckillPlaceOrderService;
import com.alibaba.fastjson.JSONObject;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * @author: kuporo
 * @since: 2025/7/10
 * @description: 监听事务消息
 */
@Component
@Slf4j
@RocketMQTransactionListener
@RequiredArgsConstructor
public class OrderTxMessageListener implements RocketMQLocalTransactionListener {
    private final SeckillPlaceOrderService seckillPlaceOrderService;
    private final DistributedCacheService distributedCacheService;
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        TxMessage txMessage = this.getTxMessage(message);
        try {
            // 已经抛出了异常,直接回滚
            if (Boolean.TRUE.equals(txMessage.getException())){
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            seckillPlaceOrderService.saveOrderInTransaction(txMessage);
            log.info("executeLocalTransaction|秒杀订单微服务成功提交本地事务|全局事务编号:{}", txMessage.getTxNo());
            return RocketMQLocalTransactionState.COMMIT;
        }catch (Exception e){
            log.error("executeLocalTransaction|秒杀订单微服务异常回滚事务|全局事务编号:{}",txMessage.getTxNo());
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        TxMessage txMessage = this.getTxMessage(message);
        log.info("checkLocalTransaction|秒杀订单微服务查询本地事务|全局事务编号:{}", txMessage.getTxNo());
        Boolean submitTransaction = distributedCacheService.hasKey(RedisConstants.getKey(RedisConstants.ORDER_TX_KEY, txMessage.getTxNo()));
        return Boolean.TRUE.equals(submitTransaction) ? RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.UNKNOWN ;

    }

    private TxMessage getTxMessage(Message msg){
        String messageString = new String((byte[]) msg.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        String txStr = jsonObject.getString(MQConstants.MSG_KEY);
        return JSONObject.parseObject(txStr, TxMessage.class);
    }
}
